package com.iwhalecloud.uncc;

import com.iwhalecloud.uncc.model.MqConsumerConfig;
import com.iwhalecloud.uncc.mq.consumer.MqConsumerManager;
import com.iwhalecloud.uncc.mq.listener.IMqHandle;
import com.iwhalecloud.uncc.mq.listener.MqListener;
import com.iwhalecloud.uncc.mq.listener.MqQueueListener;
import com.iwhalecloud.uncc.mq.listener.MqTopicListener;
import com.iwhalecloud.uncc.mq.producer.MqProducerManager;
import com.iwhalecloud.uncc.properties.mq.ArchiverConsumerProperties;
import com.iwhalecloud.uncc.scheduler.DaemonScheduler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class UnccArchiverInitialization implements ApplicationRunner {

    @Autowired
    private MqProducerManager mqProducerManager;

    @Autowired
    private MqConsumerManager mqConsumerManager;

    @Autowired
    private ArchiverConsumerProperties archiverConsumerProperties;

    @Autowired
    private DaemonScheduler daemonScheduler;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        mqProducerManager.init();
        // 初始化配置的消费者
        List<ArchiverConsumerProperties.Consumer> list =  archiverConsumerProperties.getConsumers();
        if(CollectionUtils.isNotEmpty(list)){
            MqConsumerConfig consumerConfig = new MqConsumerConfig();
            for(ArchiverConsumerProperties.Consumer c : list){
                consumerConfig.setTopic(c.getTopic());
                consumerConfig.setTag(c.getTag());
                consumerConfig.setConsumerName(c.getGroupName());
                consumerConfig.setConsumeThreadMax(c.getConsumeThreadMax());
                consumerConfig.setConsumeThreadMin(c.getConsumeThreadMin());
                consumerConfig.setConsumeBatchSize(c.getConsumeBatchSize());
                MqListener mqListener;
                if(c.isInOrder()){
                    mqListener = new MqQueueListener();
                }else{
                    mqListener = new MqTopicListener();
                }
                // 主题作为参数传入处理类
                Map<String, Object> params = new HashMap<>();
                params.put("topic", c.getTopic());
                IMqHandle handler = (IMqHandle) Class.forName(c.getHandlerClass()).newInstance();
                handler.init(params);
                mqListener.setHandler(handler);
                mqConsumerManager.createMqConsumer(consumerConfig, mqListener);
            }
        }
        //后台开启调度 daemonScheduler
        daemonScheduler.run();
        log.info("######################DaemonScheduler is Running !############################");
    }
}
